package com.sui.bigdata.flink.sql.side.program.util;

import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import com.sui.bigdata.flink.sql.side.program.Distinct;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;

/**
 * @author YongChen
 * @date 2020/5/13 16:14
 * @description
 * @email yong_chen@sui.com
 */
public class DistinctProcessFunction  extends KeyedProcessFunction<Tuple, Tuple3<String, Integer,Integer>, String> {
    private static final long serialVersionUID = 1L;
    private static final Logger LOGGER = LoggerFactory.getLogger(DistinctProcessFunction.class);
    private static final int BF_CARDINAL_THRESHOLD = 1000000;
    private static final double BF_FALSE_POSITIVE_RATE = 0.01;

    private transient  ValueState<BloomFilter<String>> subOrderFilter;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<BloomFilter<String>> subOrderFilterDescriptor = new ValueStateDescriptor<BloomFilter<String>>("boom_filter", TypeInformation.of(new TypeHint<BloomFilter<String>>() {}));
        subOrderFilter = getRuntimeContext().getState(subOrderFilterDescriptor);
    }

    @Override
    public void processElement(Tuple3<String, Integer,Integer> value, Context ctx, Collector<String> out) throws Exception {
        BloomFilter<String> bloomFilter = subOrderFilter.value();
        if (null == bloomFilter){
            bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charset.forName("GBK")), BF_CARDINAL_THRESHOLD, BF_FALSE_POSITIVE_RATE);
        }
        String subOrderId = value.f0;
        if (!bloomFilter.mightContain(subOrderId)) {
            bloomFilter.put(subOrderId);
            subOrderFilter.update(bloomFilter);
            out.collect(value.f0);
        }
        ctx.timerService().registerProcessingTimeTimer(tomorrowZeroTimestampMs(System.currentTimeMillis(), 8) + 1);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        long s = System.currentTimeMillis();
        subOrderFilter.update(BloomFilter.create(Funnels.stringFunnel(Charset.forName("GBK")), BF_CARDINAL_THRESHOLD, BF_FALSE_POSITIVE_RATE));
        long e = System.currentTimeMillis();
        LOGGER.info("Timer triggered & resetted Guava BloomFilter, time cost: " + (e - s));
    }


    public static long tomorrowZeroTimestampMs(long now, int timeZone) {
        return now - (now + timeZone * 3600000) % 86400000 + 86400000;
    }
}


